package com.showdor.springboot;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.*;
import java.util.concurrent.ExecutionException;

import com.github.javafaker.Faker;

public class TrafficDataProducer {

    private static final String TOPIC_NAME = "jtts-traffice-topic";
    // Kafka 服务器地址
    private static final String BOOTSTRAP_SERVERS = "192.168.139.185:32092";
    private static final Map<String, String> ROAD_MAP = new HashMap() {{
        put("1000", "京海高速0");
        put("1001", "京海高速1");
        put("1002", "京海高速2");
        put("1003", "京海高速3");
        put("1004", "京海高速4");
        put("1005", "京海高速5");
    }};
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final Producer<String, String> producer = new KafkaProducer<>(getKafkaProperties());
    static Faker faker = new Faker(new Locale("zh-CN"));

    public static void main(String[] args) throws InterruptedException {


        try {
            // 模拟过车数据
            int batch = 0;
            while (true) {
                sendVehicleData(batch);
                batch++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭 Kafka 生产者
            producer.close();
        }
    }

    private static Properties getKafkaProperties() {
        // 配置 Kafka 生产者
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return properties;
    }

    private static void sendVehicleData(int batch) throws InterruptedException, ExecutionException, JsonProcessingException {
        long startTime = System.currentTimeMillis();
        System.out.println("开始执行第" + batch + "次发送数据");
        int[][] ranges = {
                {1000, 2000},
                {2000, 3000},
                {3000, 3500},
                /*  {3500, 4000},
                 {4000, 4200},
                           {4300, 4700},
                            {4700, 4800},
                            {4800, 4900},
                            {4900, 5000},
                            {5000, 5100},
                            {5100, 5300},
                            {6100, 6800},
                            {6800, 7800},
                            {7800, 8800},
                            {8800, 9800},
                            {9800, 10800},
                            {10000, 11800},
                            {11000, 11300},
                            {11300, 11500},*/
        };
        int count = 0;
        for (int i = 0; i < ranges.length; i++) {
            for (int j = ranges[i][0]; j < ranges[i][1]; j++) {
                count++;
                Map<String, Object> vehicleData = generateVehicleData(faker, batch, j);
                String jsonData = objectMapper.writeValueAsString(vehicleData);
                ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, jsonData);

                producer.send(record, new Callback() {
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if (exception == null) {
                            // System.out.println("Sent data: " + jsonData);
                        } else {
                            //  System.err.println("Failed to send data: " + exception.getMessage());
                        }
                    }
                });
                Thread.sleep(10);
            }
            // 确保每秒发送一条数据
            Thread.sleep(500); // 休眠 1 秒
        }
        System.out.println("本批次：" + batch + "发送了" + count + "条数据，耗时" + (System.currentTimeMillis() - startTime) + "ms");
    }

    private static Map<String, Object> generateVehicleData(Faker faker, int batch, int num) {
        Map<String, Object> vehicleData = new HashMap<>();

        vehicleData.put("motorVehicleId", UUID.randomUUID().toString());
        vehicleData.put("infoKind", "1");
        vehicleData.put("sourceId", faker.number().digits(3));
        vehicleData.put("sbbh", faker.number().digits(15));
        vehicleData.put("tollgateId", faker.number().digits(15));
        vehicleData.put("deviceId", faker.number().digits(15));
        vehicleData.put("sbmc", faker.lorem().word() + "卡口相机");
        vehicleData.put("storageUrl1", "http://example.com/storage1");
        vehicleData.put("storageUrl2", "http://example.com/storage2");
        vehicleData.put("storageUrl3", "");
        vehicleData.put("storageUrl4", "");
        vehicleData.put("storageUrl5", "");
        vehicleData.put("markTime", "");
        vehicleData.put("laneNo", faker.number().numberBetween(1, 5));
        vehicleData.put("plateClass", "0" + (num % 6));//号牌类型 大车
        //vehicleData.put("plateClass", "0");//号牌类型 小车
        vehicleData.put("vehicleClass", getVehicleClass(num));//车辆类型 TODO 当前kafka过来的数据时没有，上线后需要加上
        vehicleData.put("plateColor", faker.number().numberBetween(0, 3));
        String mod = String.valueOf(num % 1000);
        //如果mod不够5位，在前面若干个0，补足5位
        if (mod.length() < 5) {
            mod = "00000".substring(0, 5 - mod.length()) + mod;
        }
        vehicleData.put("plateNo", "浙A"+mod); // 生成车牌号
        vehicleData.put("plateNoAttach", "");
        vehicleData.put("plateDescribe", "");
        vehicleData.put("speed", faker.number().numberBetween(30, 180));
        vehicleData.put("direction", faker.lorem().word());
        vehicleData.put("usingPropertiesCode", getUsingPropertiesCode(num));
        vehicleData.put("vehicleBrand", faker.company().name()); // 使用公司名称模拟车辆品牌
        vehicleData.put("vehicleModel", faker.lorem().word()); // 使用随机单词模拟车辆型号
        //模拟过车时间
        //startTime = startTime + count * 1000 * 60 * 32 * faker.number().numberBetween(1, 3);
        //vehicleData.put("passTime", startTime);
        vehicleData.put("passTime", System.currentTimeMillis());
        vehicleData.put("nameOfPassedRoad", faker.address().streetName());
        vehicleData.put("thirdId", UUID.randomUUID().toString());
        String dldm = faker.number().numberBetween(1000, 1005) + "";
        vehicleData.put("dldm", dldm);
        vehicleData.put("dlmc", ROAD_MAP.getOrDefault(dldm, "未知"));
        vehicleData.put("gls", faker.number().digits(4));
        vehicleData.put("ms", faker.number().digits(3));
        vehicleData.put("jd", faker.number().randomDouble(6, 0, 180));
        vehicleData.put("wd", faker.number().randomDouble(6, 0, 90));
        String glbm= String.valueOf(faker.number().numberBetween(10000, 10010));
        vehicleData.put("glbm",glbm );
        vehicleData.put("glbmmc","管理部门"+glbm);
        vehicleData.put("fxlx", faker.number().numberBetween(1, 2));
        vehicleData.put("checkInfo", "未配置集指设备");
        vehicleData.put("upStatus", "0");
        vehicleData.put("jkdbh", faker.number().numberBetween(12345678900L, 12345678910L));
        //vehicleData.put("jkdbh", "1234567890" + batch % faker.number().numberBetween(2, 9));

        vehicleData.put("kkbjlx", faker.number().numberBetween(0, 6));
        vehicleData.put("createTime", System.currentTimeMillis());
        return vehicleData;
    }

    /**
     * 生成车牌号
     */
    private static String generatePlateNumber(Faker faker) {
        // 生成一个随机的车牌号，例如 "浙A12345"
        return faker.regexify("[浙京津沪渝冀豫云辽黑湘皖鲁新苏浙赣鄂桂甘晋蒙陕吉闽贵粤青藏川宁琼][A-Z]\\d{5,6}");
    }

    /**
     * 车辆类型
     *
     * @return
     */
    private static String getVehicleClass(Integer num) {
        //从以逗号分隔的字符串中随机选择一个元素
        String[] elements = {
                "B11", "B12", "B13", "B14", "B15", "B16", "B17", "B18", "B19", "B1A", "B1B", "B1C", "B1D", "B1E", "B1F", "B1G", "B1H", "B1J", "B1K", "B1L", "B1U", "B1V", "B1W",
                "B21", "B22", "B23", "B24", "B25", "B26", "B27", "B28", "B29", "B2A", "B2B", "B2C", "B2D", "B2E", "B2F", "B2G", "B2H", "B2J", "B2K", "B2L", "B2U", "B2V", "B2W",
                "B31", "B32", "B33", "B34", "B35", "B36", "B37", "B38", "B39", "B3C", "B3D", "B3E", "B3F", "B3G", "B3H", "B3J", "B3K", "B3L", "B3U", "B3V", "B3W",
                "B41", "B42", "B43", "B44", "B45", "B46", "B47", "B48", "B49", "B4C", "B4D", "B4E", "B4F", "B4G", "B4H", "B4J", "B4K", "B4L", "B4U", "B4V", "B4W",
                "D11", "D12", "G11", "G12", "G13", "G14", "G15", "G16", "G17", "G18", "G19", "G1A", "G1B", "G1C", "G1D", "G1E", "G1F",
                "G21", "G22", "G23", "G24", "G25", "G26", "G27", "G28", "G29", "G2A", "G2B", "G2C", "G2D", "G2E", "G2F",
                "G31", "G32", "G33", "G34", "G35", "G36", "G37", "G38", "G3A", "G3B", "G3C", "G3D", "G3E", "G3F",
                "G41", "G42", "G43", "G44", "G45", "G46", "G48", "G4A", "G4B", "G4C", "G4D", "G4E", "G4F", "H11", "H12", "H13", "H14", "H15", "H16", "H17", "H18", "H19", "H1A", "H1B", "H1C", "H1D", "H1E", "H1F", "H1G", "H1J",
                "H21", "H22", "H23", "H24", "H25", "H26", "H27", "H28", "H29", "H2A", "H2B", "H2C", "H2D", "H2E", "H2F", "H2G", "H2J",
                "H31", "H32", "H33", "H34", "H35", "H37", "H38", "H39", "H3A", "H3B", "H3C", "H3D", "H3F", "H3G", "H3H", "H3J",
                "H41", "H42", "H43", "H44", "H45", "H46", "H47", "H4A", "H4B", "H4C", "H4F", "H4G", "H4H", "H4J",
                "H51", "H52", "H53", "H54", "H55", "H5B", "H5C",
                "J11", "J12", "J13",
                "K11", "K12", "K13", "K14", "K15", "K16", "K17", "K18", "K1A",
                "K21", "K22", "K23", "K24", "K25", "K26", "K27", "K28", "K2A",
                "K31", "K32", "K33", "K34", "K38", "K39", "K3A",
                "K41", "K42", "K43", "K49", "K4A",
                "M11", "M12", "M13", "M14", "M15"
        };
        String randomElement = elements[num % 6];
        return randomElement;
    }

    private static String getUsingPropertiesCode(Integer num) {
        return ("ABCDEFGHIJKLMNOPQRSTU").substring(num % 6, num % 6 + 1);
    }

    public static void createTopic(String bootstrapServers, String topicName, int numPartitions, short replicationFactor) {
        // 配置AdminClient
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        // 创建AdminClient
        try (AdminClient adminClient = AdminClient.create(props)) {
            // 定义新主题
            NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);

            // 创建主题
            adminClient.createTopics(Collections.singleton(newTopic)).all().get();

            System.out.println("Topic '" + topicName + "' created successfully with " + numPartitions + " partitions.");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 检查主题是否存在，如果存在则删除
     */
    public static void deleteTopicIfExists(String bootstrapServers, String topicName) {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        try (AdminClient adminClient = AdminClient.create(props)) {
            // 获取所有主题
            ListTopicsResult topicsResult = adminClient.listTopics();
            Set<String> topics = topicsResult.names().get();

            // 检查主题是否存在
            if (topics.contains(topicName)) {
                System.out.println("Topic '" + topicName + "' exists. Deleting it...");

                // 删除主题
                DeleteTopicsResult deleteResult = adminClient.deleteTopics(Collections.singleton(topicName));
                deleteResult.all().get(); // 等待删除完成

                System.out.println("Topic '" + topicName + "' deleted successfully.");
            } else {
                System.out.println("Topic '" + topicName + "' does not exist.");
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

}